﻿namespace AbpPlusPlus.RabbitMq
{
    #region

    using System;
    using System.Collections.Concurrent;
    using System.Collections.Generic;
    using System.Linq;
    using System.Net.Sockets;
    using System.Reflection;
    using System.Text;
    using System.Threading.Tasks;

    using Abp;
    using Abp.Dependency;
    using Abp.Events.Bus;
    using Abp.Events.Bus.Factories;
    using Abp.Events.Bus.Handlers;
    using Abp.Extensions;
    using Abp.Threading.Extensions;

    using AbpPlusPlus.RabbitMq.Configuration;
    using AbpPlusPlus.RabbitMq.Factories.Internals;

    using Castle.Core.Logging;

    using Newtonsoft.Json;

    using Polly;

    using RabbitMQ.Client;
    using RabbitMQ.Client.Events;
    using RabbitMQ.Client.Exceptions;

    #endregion

    /// <summary>
    ///     支持Action的事件处理器
    /// </summary>
    /// <typeparam name="TEventData"></typeparam>
    internal class ActionEventHandler<TEventData> : IEventHandler<TEventData>
        where TEventData : IEventData
    {
        public ActionEventHandler(Action<TEventData> handler)
        {
            this.Action = handler;
        }

        /// <summary>
        ///     定义Action的引用，并通过构造函数传参初始化
        /// </summary>
        public Action<TEventData> Action { get; }

        /// <summary>
        ///     调用具体的Action来处理事件逻辑
        /// </summary>
        /// <param name="eventData"></param>
        public void HandleEvent(TEventData eventData)
        {
            this.Action(eventData);
        }
    }

    public class RabbitMqEventBus : IEventBus
    {
        private const string BROKER_NAME = "abp_event_bus";
        
        private readonly ConcurrentDictionary<Type, List<IEventHandlerFactory>> _handlerFactories;

        private readonly IRabbitMqProvider _rabbitMqProvider;

        private readonly IAbpRabbitMqModuleConfiguration _rabbitMqModuleConfiguration;

        private IModel _consumerChannel;

        private string _queueName;

        public RabbitMqEventBus(IRabbitMqProvider rabbitMqProvider, IAbpRabbitMqModuleConfiguration rabbitMqModuleConfiguration)
        {
            this._rabbitMqProvider = rabbitMqProvider;
            this._rabbitMqModuleConfiguration = rabbitMqModuleConfiguration;
            this._handlerFactories = new ConcurrentDictionary<Type, List<IEventHandlerFactory>>();
            this._consumerChannel = this.CreateConsumerChannel();
            this.Logger = NullLogger.Instance;
        }

        public IIocManager IocManager { get; set; }

        public ILogger Logger { get; set; }

        public IDisposable Register<TEventData>(Action<TEventData> action)
            where TEventData : IEventData
        {
            return this.Register(typeof(TEventData), new ActionEventHandler<TEventData>(action));
        }

        public IDisposable Register<TEventData>(IEventHandler<TEventData> handler)
            where TEventData : IEventData
        {
            return this.Register(typeof(TEventData), handler);
        }

        public IDisposable Register<TEventData, THandler>()
            where TEventData : IEventData where THandler : IEventHandler<TEventData>, new()
        {
            return this.Register(typeof(TEventData), new TransientEventHandlerFactory<THandler>());
        }

        public IDisposable Register(Type eventType, IEventHandler handler)
        {
            return this.Register(eventType, new SingleInstanceHandlerFactory(handler));
        }

        public IDisposable Register<TEventData>(IEventHandlerFactory handlerFactory)
            where TEventData : IEventData
        {
            return this.Register(typeof(TEventData), handlerFactory);
        }

        public IDisposable Register(Type eventType, IEventHandlerFactory handlerFactory)
        {
            this.GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Add(handlerFactory));

            using (var channel = this._rabbitMqProvider.CreateModel())
            {
                channel.QueueBind(queue: this._queueName, exchange: BROKER_NAME, routingKey: eventType.Name);
            }

            return new FactoryUnregistrar(this, eventType, handlerFactory);
        }

        public void Trigger<TEventData>(TEventData eventData)
            where TEventData : IEventData
        {
            this.Trigger((object)null, eventData);
        }

        public void Trigger<TEventData>(object eventSource, TEventData eventData)
            where TEventData : IEventData
        {
            this.Trigger(typeof(TEventData), eventSource, eventData);
        }

        public void Trigger(Type eventType, IEventData eventData)
        {
            this.Trigger(eventType, null, eventData);
        }

        public void Trigger(Type eventType, object eventSource, IEventData eventData)
        {
            var exceptions = new List<Exception>();

            // 实际上需要在这里，触发rabbit 事件
            this.TriggerHandlingException(eventType, eventSource, eventData, exceptions);

            if (exceptions.Any())
            {
                if (exceptions.Count == 1)
                {
                    exceptions[0].ReThrow();
                }

                throw new AggregateException(
                    "More than one error has occurred while triggering the event: " + eventType,
                    exceptions);
            }
        }

        public Task TriggerAsync<TEventData>(TEventData eventData)
            where TEventData : IEventData
        {
            return this.TriggerAsync((object)null, eventData);
        }

        public Task TriggerAsync<TEventData>(object eventSource, TEventData eventData)
            where TEventData : IEventData
        {
            /*
#if NET46
            ExecutionContext.SuppressFlow();
#endif
*/
            var task = Task.Factory.StartNew(
                () =>
                    {
                        try
                        {
                            this.Trigger(eventSource, eventData);
                        }
                        catch (Exception ex)
                        {
                            this.Logger.Warn(ex.ToString(), ex);
                        }
                    });

            /*
#if NET46
            ExecutionContext.RestoreFlow();
#endif
*/

            return task;
        }

        public Task TriggerAsync(Type eventType, IEventData eventData)
        {
            return this.TriggerAsync(eventType, null, eventData);
        }

        public Task TriggerAsync(Type eventType, object eventSource, IEventData eventData)
        {
            var task = Task.Factory.StartNew(
                () =>
                    {
                        try
                        {
                            this.Trigger(eventType, eventSource, eventData);
                        }
                        catch (Exception ex)
                        {
                            this.Logger.Warn(ex.ToString(), ex);
                        }
                    });

            return task;
        }

        public void Unregister<TEventData>(Action<TEventData> action)
            where TEventData : IEventData
        {
            Check.NotNull(action, nameof(action));

            this.GetOrCreateHandlerFactories(typeof(TEventData)).Locking(
                factories =>
                    {
                        factories.RemoveAll(
                            factory =>
                                {
                                    var singleInstanceFactory = factory as SingleInstanceHandlerFactory;
                                    if (singleInstanceFactory == null)
                                    {
                                        return false;
                                    }

                                    var actionHandler =
                                        singleInstanceFactory.HandlerInstance as ActionEventHandler<TEventData>;
                                    if (actionHandler == null)
                                    {
                                        return false;
                                    }

                                    return actionHandler.Action == action;
                                });
                    });

            this.Unbind(typeof(TEventData));
        }

        public void Unregister<TEventData>(IEventHandler<TEventData> handler)
            where TEventData : IEventData
        {
            this.Unregister(typeof(TEventData), handler);
        }

        public void Unregister(Type eventType, IEventHandler handler)
        {
            this.GetOrCreateHandlerFactories(eventType).Locking(
                factories =>
                    {
                        factories.RemoveAll(
                            factory => factory is SingleInstanceHandlerFactory
                                       && (factory as SingleInstanceHandlerFactory).HandlerInstance == handler);
                    });
            this.Unbind(eventType);
        }

        public void Unregister<TEventData>(IEventHandlerFactory factory)
            where TEventData : IEventData
        {
            this.Unregister(typeof(TEventData), factory);
        }

        public void Unregister(Type eventType, IEventHandlerFactory factory)
        {
            this.GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Remove(factory));
            this.Unbind(eventType);
        }

        public void UnregisterAll<TEventData>()
            where TEventData : IEventData
        {
            this.UnregisterAll(typeof(TEventData));
        }

        public void UnregisterAll(Type eventType)
        {
            this.GetOrCreateHandlerFactories(eventType).Locking(factories => factories.Clear());

            this.Unbind(eventType);
        }

        private void Unbind(Type eventType)
        {
            using (var channel = this._rabbitMqProvider.CreateModel())
            {
                channel.QueueUnbind(queue: this._queueName, exchange: BROKER_NAME, routingKey: eventType.Name);
            }
        }

        private IModel CreateConsumerChannel()
        {
            if (!this._rabbitMqProvider.IsConnected)
            {
                this._rabbitMqProvider.TryConnect();
            }

            var channel = this._rabbitMqProvider.CreateModel();

            channel.ExchangeDeclare(exchange: BROKER_NAME, type: "direct");

            // this._rabbitMqModuleConfiguration.EventQueueName
            this._queueName = channel.QueueDeclare(
                queue: this._rabbitMqModuleConfiguration.EventQueueName,
                exclusive:false).QueueName;

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
                {
                    var eventName = ea.RoutingKey;
                    var message = Encoding.UTF8.GetString(ea.Body);

                    var eventType = this._handlerFactories.Keys.Single(c => c.Name == eventName);
                    var integrationEvent = JsonConvert.DeserializeObject(message, eventType);
                    this.ProcessEvent(eventType, integrationEvent as IEventData);
                };

            channel.BasicConsume(queue: this._queueName, autoAck: true, consumer: consumer);

            channel.CallbackException += (sender, ea) =>
                {
                    this._consumerChannel.Dispose();
                    this._consumerChannel = this.CreateConsumerChannel();
                };

            return channel;
        }

        private List<IEventHandlerFactory> GetOrCreateHandlerFactories(Type eventType)
        {
            return this._handlerFactories.GetOrAdd(eventType, type => new List<IEventHandlerFactory>());
        }

        private IEnumerable<EventTypeWithEventHandlerFactories> GetHandlerFactories(Type eventType)
        {
            var handlerFactoryList = new List<EventTypeWithEventHandlerFactories>();

            foreach (var handlerFactory in this._handlerFactories.Where(
                hf => ShouldTriggerEventForHandler(eventType, hf.Key)))
            {
                handlerFactoryList.Add(
                    new EventTypeWithEventHandlerFactories(handlerFactory.Key, handlerFactory.Value));
            }

            return handlerFactoryList.ToArray();
        }

        private static bool ShouldTriggerEventForHandler(Type eventType, Type handlerType)
        {
            //Should trigger same type
            if (handlerType == eventType)
            {
                return true;
            }

            //Should trigger for inherited types
            if (handlerType.IsAssignableFrom(eventType))
            {
                return true;
            }

            return false;
        }

        private void ProcessEvent(Type eventType, IEventData eventData)
        {
            foreach (var handlerFactories in this.GetHandlerFactories(eventType))
            {
                foreach (var handlerFactory in handlerFactories.EventHandlerFactories)
                {
                    var eventHandler = handlerFactory.GetHandler();

                    try
                    {
                        if (eventHandler == null)
                        {
                            throw new Exception(
                                $"Registered event handler for event type {handlerFactories.EventType.Name} does not implement IEventHandler<{handlerFactories.EventType.Name}> interface!");
                        }

                        var handlerType = typeof(IEventHandler<>).MakeGenericType(handlerFactories.EventType);

                        var method = handlerType.GetMethod("HandleEvent", new[] { handlerFactories.EventType });

                        method.Invoke(eventHandler, new object[] { eventData });
                    }
                    catch (TargetInvocationException ex)
                    {
                        //exceptions.Add(ex.InnerException);
                    }
                    catch (Exception ex)
                    {
                        //exceptions.Add(ex);
                    }
                    finally
                    {
                        handlerFactory.ReleaseHandler(eventHandler);
                    }
                }
            }

            //Implements generic argument inheritance. See IEventDataWithInheritableGenericArgument
            if (eventType.GetTypeInfo().IsGenericType && eventType.GetGenericArguments().Length == 1
                && typeof(IEventDataWithInheritableGenericArgument).IsAssignableFrom(eventType))
            {
                var genericArg = eventType.GetGenericArguments()[0];
                var baseArg = genericArg.GetTypeInfo().BaseType;
                if (baseArg != null)
                {
                    var baseEventType = eventType.GetGenericTypeDefinition().MakeGenericType(baseArg);
                    var constructorArgs = ((IEventDataWithInheritableGenericArgument)eventData).GetConstructorArgs();
                    var baseEventData = (IEventData)Activator.CreateInstance(baseEventType, constructorArgs);
                    baseEventData.EventTime = eventData.EventTime;
                    this.Trigger(baseEventType, eventData.EventSource, baseEventData);
                }
            }
        }

        private void TriggerHandlingException(
            Type eventType,
            object eventSource,
            IEventData eventData,
            List<Exception> exceptions)
        {
            //TODO: This method can be optimized by adding all possibilities to a dictionary.

            eventData.EventSource = eventSource;

            if (!this._rabbitMqProvider.IsConnected)
            {
                this._rabbitMqProvider.TryConnect();
            }

            var policy = Policy.Handle<BrokerUnreachableException>().Or<SocketException>().WaitAndRetry(
                5,
                retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)),
                (ex, time) => { this.Logger.Warn(ex.ToString()); });

            using (var channel = this._rabbitMqProvider.CreateModel())
            {
                if (channel==null)
                {
                    return;
                }

                var eventName = eventType.Name;

                channel.ExchangeDeclare(exchange: BROKER_NAME, type: "direct");

                var message = JsonConvert.SerializeObject(eventData);
                var body = Encoding.UTF8.GetBytes(message);

                policy.Execute(
                    () =>
                        {
                            channel.BasicPublish(
                                exchange: BROKER_NAME,
                                routingKey: eventName,
                                basicProperties: null,
                                body: body);
                        });
            }
        }

        private class EventTypeWithEventHandlerFactories
        {
            public EventTypeWithEventHandlerFactories(Type eventType, List<IEventHandlerFactory> eventHandlerFactories)
            {
                this.EventType = eventType;
                this.EventHandlerFactories = eventHandlerFactories;
            }

            public Type EventType { get; }

            public List<IEventHandlerFactory> EventHandlerFactories { get; }
        }
    }
}